package com.shujia.streaming

import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo01WCntOnStreaming {
  def main(args: Array[String]): Unit = {
    // 构建SparkContext
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo01WCntOnStreaming")

    /**
     * WARN StreamingContext: spark.master should be set as local[n],
     * n > 1 in local mode if you have receivers to get data
     * , otherwise Spark jobs will not get resources to process the received data.
     *
     * 在使用StreamingContext的时候需要至少指定两个线程
     * 因为接收数据需要一直占用一个线程，如果没有两个线程以上
     * 则接受到的数据无法进行处理
     */
    conf.setMaster("local[2]")
    val sc: SparkContext = new SparkContext(conf)

    // 构建StreamingContext
    /**
     * 需要传入两个参数：sc、时间
     * 时间相当于控制SparkStreaming每个批次的大小
     * SparkStreaming本质上是伪实时，微批处理
     * 如下：则表示5s一个批次
     * 将5s钟的数据封装成一个batch，然后产生一个job进行处理
     */
    val ssc: StreamingContext = new StreamingContext(sc, Durations.seconds(5))

    /**
     * 通过nc命令来模拟消息队列，构建一个流
     * yum install nc
     *
     * nc -lk 8888
     *
     * DStream是SparkStreaming中的编程模型，主要用于流处理
     */

    val lineDS: DStream[String] = ssc.socketTextStream("master", 8888)

    // 统计每个单词的数量
    /**
     * 每个批次都是独立的，每次的计算结果互不影响
     * 每次的计算结果可以看成某一时刻SparkStreaming任务运行的状态
     *
     */
    lineDS
      .flatMap(_.split(","))
      .map(word => (word, 1))
      .reduceByKey(_ + _)
      .print()

    // 启动流任务
    ssc.start()
    ssc.awaitTermination() // 等待停止
    ssc.stop() // 停止任务

  }

}
